// Copyright (C) 2008 Taylor L. Riche <riche@cs.utexas.edu>
//  
// This program is free software; you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation; either version 2 of the License, or
// (at your option) any later version.
// 
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
// 
// You should have received a copy of the GNU General Public License
// along with this program; if not, write to the Free Software 
// Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
//

#include "file_reader.hpp"

using namespace std;

FileReader::FileReader(const char * ignore, void * vp) : sysReady(false), inFile(IN_FILE.c_str()), outFile(OUT_FILE.c_str())
{
  pthread_attr_t attr;
  pthread_attr_init(&attr);
  pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
  if(pthread_create(&coreThread, &attr, startDevice,(void*) this)) {
    std::cerr << "pthread_create failed" << strerror(errno) << std::endl;
    std::exit(1);
  }
}

FileReader::~FileReader() 
{
  if(inFile.is_open()) inFile.close();
  if(outFile.is_open()) outFile.close();
}

void 
FileReader::process() 
{
  //std::cerr << "filereader:: about to wait" << std::endl;
  {
    Synchronized::Scope mutex(this);
    while(!sysReady) {
      wait();
    }
  }
  boost::shared_ptr<string> line;
  if(inFile.is_open()) {
    while(!inFile.eof()) {
      std::string * tempString = new std::string;
      std::getline(inFile, *tempString);
      line.reset(tempString);
      StringWrapper * temp = new StringWrapper();
      temp->token = false;
      temp->coreString = line;
      //std::cerr << "Just read: " << *line << std::endl;
      std::pair<lagniappe::CoreOp *, uint32_t> * dest = destVec[3];
      assert(dest != NULL);
      assert(dest->first != NULL);
      lagniappe::RCon * newRCon = new lagniappe::RCon(dest->first, static_cast<lagniappe::RData *>(temp), dest->second, sizeof(temp));
      dest->first->run(newRCon);
    }
    StringWrapper * done = new StringWrapper();
    done->token = true;
    std::pair<lagniappe::CoreOp *, uint32_t> * dest = destVec[3];
    assert(dest != NULL);
    assert(dest->first != NULL);
    lagniappe::RCon * newRCon = new lagniappe::RCon(dest->first, static_cast<lagniappe::RData *>(done), dest->second, sizeof(done));
    dest->first->run(newRCon);
  }
  
  return;
}

void 
FileReader::run(lagniappe::RCon * rcon)
{
  StringWrapper * req = static_cast<StringWrapper *>(rcon->getRData());
  assert(req != NULL);
  if(req->token) {
    outFile.close();
    exit(0);
  }
  else {
    boost::shared_ptr<string> line = req->coreString;
    outFile << *line << std::endl;
    std::cout << *line << std::endl;
    delete req;
    delete rcon;
  }
  return;
}

void 
FileReader::start()
{
  Synchronized::Scope mutex(this);
  sysReady = true;
  notifyAll();
  return;
}


#ifdef __cplusplus
extern "C" {
#endif /* __cplusplus */

void * 
startDevice(void * arg)
{

  FileReader * fr = static_cast<FileReader *>(arg);
  fr->process();
  return NULL;
}

#ifdef __cplusplus
}
#endif /* __cplusplus */
